package mr

import (
	"fmt"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"regexp"
	"sync"
	"time"
)

const (
	TASK_NONE   = iota // 工作不受理
	TASK_MAP           // 系统执行Map任务阶段
	TASK_REDUCE        // 系统执行Reduce任务阶段
	TASK_OK            // 系统工作完毕
)

// Master节点的工作组件
// [1] 访问控制锁
// [2] Map/Reduce的任务等待队列
// [3] 定时器监督模块
type Master struct {
	// Your definitions here.
	FileList  []string      // 需要解析的最初输入文件
	Timeout   time.Duration // 定时器时间
	ReduceCnt int           // 需要执行的nReduce数目

	// 任务执行Runtime模块
	FilePools  OutputFilePool // 结果文件标记池
	TaskMan    TaskManager    // 任务管理器
	mu         *sync.Mutex    // 变量访问控制锁
	MRWorkStat int            // 当前系统的工作状态
}

// 定时器监督器 - 启动监督线程
func (m *Master) taskSupervis(task *Task) {
	go func(taskType, taskID, taskRunID int) {
		MrLoger.Printf("TaskSuperVis: TaskID:%v,TaskRunID:%v", taskID, taskRunID)
		tst := time.NewTimer(m.Timeout)
		<-tst.C // 阻塞等待数据
		// 检查系统整体状态是否已经变化
		if m.MRWorkStat == TASK_OK || m.MRWorkStat == TASK_NONE {
			return
		}
		MrLoger.Printf("SupervisCheck: TaskID:%v,TaskRunID:%v", taskID, taskRunID)
		// 如果任务状态符合
		if taskType == m.MRWorkStat {
			m.mu.Lock()
			defer m.mu.Unlock()
			// 释放任务记录资源
			if !m.TaskMan.isComplete(taskID) {
				// 任务回退
				m.TaskMan.PushBackRunTask(taskRunID)
			} else {
				// 任务已经完成 - 删除该taskRunID任务即可
				m.TaskMan.DeleteRunTask(taskRunID)
			}
		}
	}(task.TaskType, task.TaskID, task.TaskRunID)
}

// 获取一个需要执行的任务 - 不带锁
func (m *Master) assignOneTask() *Task {
	task := m.TaskMan.PubTask()
	// MrLoger.Printf("AssignOneTask: TaskID:%v,TaskRunID:%v", task.TaskID, task.TaskRunID)
	return task
}

// 初始化Map工作
func (m *Master) initMap() {
	MrLoger.Print("initMap() Begin")
	// 要求状态不被打扰
	m.MRWorkStat = TASK_NONE

	MrLoger.Print("initMap() Store Tasks")
	// 任务预装载
	m.TaskMan.Init_Manager(TASK_MAP)
	for _, file := range m.FileList {
		tagfiles := []TagFile{GenFileNoTag(file)}
		m.TaskMan.Init_LoadTask(TASK_MAP, tagfiles)
	}
	m.FilePools.Init_Pool() //结果文件池初始化

	// 状态重置
	m.MRWorkStat = TASK_MAP
}

// 初始化Reduce工作 - 函数外必须加锁
func (m *Master) initReduce() {
	MrLoger.Print("initReduce() Begin")
	// 要求状态不被打扰
	m.MRWorkStat = TASK_NONE

	MrLoger.Print("initReduce() MidFile Convert and Store Tasks")
	// 中间文件转换和任务预装载
	taglis := m.FilePools.GetTags_Y()
	MrLoger.Printf("taglis: %v\n", taglis)
	m.TaskMan.Init_Manager(TASK_REDUCE) // 任务管理器初始化
	for _, tagY := range taglis {
		tagfilesY := m.FilePools.GetTagFiles_Y(tagY)
		MrLoger.Printf("tag[%v] files=> %v", tagY, tagfilesY)
		m.TaskMan.Init_LoadTask(TASK_REDUCE, tagfilesY)
	}
	m.FilePools.Init_Pool() //结果文件池初始化

	// 状态重置
	m.MRWorkStat = TASK_REDUCE
}

// 初始化
func (m *Master) setup(files []string, nReduce int) {
	MrLoger.Print("setup() Begin")
	// 全局基本变量初始化
	m.MRWorkStat = TASK_NONE // 首先标记Master状态
	m.mu = &sync.Mutex{}     // 锁初始化仅1次
	m.FileList = files
	m.Timeout = time.Second * 10
	m.ReduceCnt = nReduce
	m.TaskMan = TaskManager{
		Task_All_Cnt:      -1,
		Task_Complete_Cnt: 0,
	}

	// 启动Map初始化
	m.initMap()

	// 启动进度推进函数
	go m.processChecker()
}

// 文件转正程序
func (m *Master) finalFileOutput() []string {
	// 此时已经设置TASK_OK且已经上锁
	// 文件验收开始
	resFiles := []string{}
	for i, tf := range m.FilePools.FileList {
		fmt.Printf("Final Mid File Check : [%v]\n", i)
		MrLoger.Printf("Final Mid File Check : [%v]\n", i)
		outfile := fmt.Sprintf("mr-out-%v", tf.GetTag_Y())
		MrLoger.Printf("Prepare OutFile : %v", outfile)
		mf, err := ReadMidFile(tf.FileName)
		if err != nil {
			fmt.Printf("Final File Read Err: %v\n", err)
			MrLoger.Printf("Final File Read Err: %v\n", err)
			continue
		}
		info := ""
		for _, kvl := range mf.KVsMap {
			info += fmt.Sprintf("%v %v\n", kvl.Key, kvl.ValueList[0])
		}
		// 写入新文件
		file, err := os.OpenFile(outfile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
		if err != nil {
			fmt.Printf("Final File Open Err: %v\n", err)
			continue
		}
		defer file.Close()
		_, err = file.Write([]byte(info))
		if err != nil {
			fmt.Printf("Final File Write Err: %v\n", err)
			continue
		}
		resFiles = append(resFiles, outfile)
	}
	return resFiles
}

// 删除所有的*-MR-*.json中间文件
func delAllMidFile() {
	// 删除目录下的所有中间文件
	delFileList, err := getDeleteFileList()
	if err != nil {
		MrLoger.Printf("[F]Remove Err %v", err)
	}
	del := true
	if del {
		for _, filename := range delFileList {
			os.Remove(filename)
			MrLoger.Printf("[F->]Remove %v", filename)
		}
	}
}

// 获取需要删除的文件的列表
func getDeleteFileList() ([]string, error) {
	lis, err := os.ReadDir(".")
	if err != nil {
		return nil, err
	}
	re := regexp.MustCompile(`\d+-[MR]-\d+.json`)
	mfunc := func(s string) bool { return re.MatchString(s) }
	fileList := make([]string, 0)
	for _, de := range lis {
		if !de.IsDir() && mfunc(de.Name()) {
			fileList = append(fileList, de.Name())
		}
	}
	return fileList, nil
}

// 进程检测推进函数
func (m *Master) processChecker() {
	MrLoger.Print("processChecker() Called")
	for {
		// 睡眠
		time.Sleep(time.Second)
		// 检查进度
		if m.MRWorkStat == TASK_MAP && m.TaskMan.Done() {
			MrLoger.Print("processChecker() Map is OK!")
			// 检测Map任务完成
			m.mu.Lock()
			m.initReduce() // 初始化Reduce任务
			m.mu.Unlock()
		} else if m.MRWorkStat == TASK_REDUCE && m.TaskMan.Done() {
			MrLoger.Print("processChecker() Reduce is OK!")
			// 检测Reduce任务完成
			m.mu.Lock()
			// 文件转正程序
			out := m.finalFileOutput()
			MrLoger.Printf("MapReduceLite : %v\n", out)
			delAllMidFile() //删除所有的中间文件
			// 任务结束标志 - 全局唯一的设置TASK_OK的位置
			m.MRWorkStat = TASK_OK
			MrLoger.Print("[OVER] MapReduce All Tasks Over")
			m.mu.Unlock()
			return // 任务完成函数退出
		}
	}
}

// Your code here -- RPC handlers for the worker to call.
// 这里定义的函数是Master的RPC函数 被worker远程调用
// 函数需要使用到的传递参数需要在rpc.go中定义

// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
// func (m *Master) Example(args *ExampleArgs, reply *ExampleReply) error {
// 	reply.Y = args.X + 1
// 	return nil
// }

// RPC - ReportTask
func (m *Master) ReportTask(req *ReportRequest, resp *ReportRespone) error {
	MrLoger.Print("ReportTask() RPC Called")
	// Master忙 - 等待后访问
	if m.MRWorkStat == TASK_NONE {
		resp.Tops = OPS_WAIT // 等待后重新汇报
		return nil
	}

	// 所有任务都结束 - 要求Worker结束
	if m.MRWorkStat == TASK_OK {
		resp.Tops = OPS_STOP // Worker节点停止工作
		return nil
	}

	// 重置标记 - 默认要求重新汇报
	resp.Tops = OPS_WAIT
	// 递交的已经完成的任务
	okTask := req.WorkTask

	// 任务类型不合则不处理
	if okTask.TaskType != m.MRWorkStat {
		return nil
	}

	// 任务完成检查
	runID := okTask.TaskRunID
	MrLoger.Printf("ReportTask() RPC TaskRunID:%v", runID)
	m.mu.Lock() // 锁住
	result := m.TaskMan.CompleteTask(runID)
	if result {
		MrLoger.Printf("Task Complete: TaskRunID:%v", runID)
		// 任务结果有效 允许结果文件写入文件池
		for _, tf := range okTask.TaskFile {
			m.FilePools.InsertTagFile(&tf)
		}
	}
	m.mu.Unlock()

	// 汇报完成
	resp.Tops = OPS_NONE
	return nil
}

// RPC - AssignTask
func (m *Master) AssignTask(req *AssignRequest, resp *AssignRespone) error {
	MrLoger.Print("AssignTask() RPC Called")
	// Master忙 - 等待后访问
	if m.MRWorkStat == TASK_NONE {
		resp.Tops = OPS_WAIT
		return nil
	}

	// 所有任务都结束 - 要求Worker结束
	if m.MRWorkStat == TASK_OK {
		resp.Tops = OPS_STOP
		return nil
	}

	// 重置标记
	resp.Tops = OPS_NONE

	if req.Tops == OPS_REQUEST {
		m.mu.Lock()
		defer m.mu.Unlock()
		resp.ReduceCnt = m.ReduceCnt

		ok := false
		// 设置标记
		if m.MRWorkStat == TASK_MAP {
			resp.Tops = OPS_M
			ok = true
		} else if m.MRWorkStat == TASK_REDUCE {
			resp.Tops = OPS_R
			ok = true
		}

		if ok {
			// 请求获取一个任务
			newTask := m.assignOneTask()
			if newTask == nil {
				resp.Tops = OPS_NONE
			} else {
				resp.WorkTask = *newTask
				MrLoger.Printf("AssignOneTask: TaskID:%v,TaskRunID:%v", newTask.TaskID, newTask.TaskRunID)
				// 启动计时监督器
				m.taskSupervis(newTask)
			}
		}
	}
	return nil
}

// start a thread that listens for RPCs from worker.go
func (m *Master) server() {
	// 注册RPC函数
	rpc.Register(m)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := masterSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	// 一个新的master线程用于监听RPC调用请求
	go http.Serve(l, nil)
}

// main/mrmaster.go calls Done() periodically to find out
// if the entire job has finished.
// 这里需要提供一种在master端检测是否已经完成所有的工作的功能
func (m *Master) Done() bool {
	return m.MRWorkStat == TASK_OK
}

// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce tasks to use.
func MakeMaster(files []string, nReduce int) *Master {
	m := Master{}

	// Your code here.
	Init_MasterLogger() // 初始化Master节点日志
	MrLoger.Printf("MakeMaster(nReduce:%v) files:%v", nReduce, files)
	m.setup(files, nReduce)

	m.server()
	return &m
}
